#include <algorithm>
#include "BigQ.h"

static int numBigQ = 0;

void* BigQThread (void* arg) {
	// read data from in pipe sort them into runlen pages

    BigQInfo *mybqi=new BigQInfo();
    mybqi=(BigQInfo*)arg;
    Pipe *in=mybqi->in;
    Pipe *out=mybqi->out;
    OrderMaker sortorder=mybqi->order;
    int runlen=mybqi->runLength;
    File midFile=mybqi->midFile;
    int pagesInRun[10000] = {0};

    char filename[1000];
    int whichBigQ = numBigQ;
    sprintf(filename, "./midFile_%d.bin", numBigQ++);
    Page dbPage;
    int pageCount = 0;                              // check whether we have read in runlen pages
    int runCount = 0;                                 // # of runs in the midfile, which equals to the k in k-way merge

    Record temp;
    Record* firstRec = new Record;

    vector<Record*> sortPage;

    midFile.Open(0, filename);

    int recCount = 0;
    int pCount = 0;
    //Schema* s = new Schema ("catalog", "orders");
    while(in->Remove(&temp)) {
        recCount++;
        if(dbPage.Append(&temp) == 0) {
            pCount++;
            pageCount++;
            while(dbPage.GetFirst(firstRec) == 1) {
                Record* tempRec = new Record;
                tempRec->Consume(firstRec);
                sortPage.push_back(tempRec);
            }
            dbPage.EmptyItOut();
            dbPage.Append(&temp);
        }
        if(pageCount == runlen) {
            runCount++;
            std::sort(sortPage.begin(), sortPage.end(), Comparator(sortorder, false));
            Page tempPage;
            for(int i = 0; i < sortPage.size(); i++) {
                if(tempPage.Append((sortPage[i])) == 0) {
                    if(midFile.GetLength() == 0)
                    {
                        midFile.AddPage(&tempPage, midFile.GetLength());
                        pagesInRun[runCount - 1]++;
                    }
                    else
                    {
                        midFile.AddPage(&tempPage, midFile.GetLength() - 1);
                        pagesInRun[runCount - 1]++;
                    }
                    tempPage.EmptyItOut();
                    tempPage.Append((sortPage[i]));
                }
            }
            // there must be one page left that has not been at to the file in current run
            if(midFile.GetLength() == 0)
            {
                midFile.AddPage(&tempPage, midFile.GetLength());
                pagesInRun[runCount - 1]++;
            }
            else
            {
                midFile.AddPage(&tempPage, midFile.GetLength() - 1);
                pagesInRun[runCount - 1]++;
            }
            tempPage.EmptyItOut();

            pageCount = 0;
            
            for(int i = 0; i < sortPage.size(); i++) {
                delete sortPage[i];
            }
            sortPage.clear();
        }
    }

    // maybe there is one suspended run/page, actulally it must be one page left, has not been push into the vector
    // which means there must be one more run
    pCount++;
    while(dbPage.GetFirst(firstRec) == 1) {
        Record* tempRec = new Record;
        tempRec->Consume(firstRec);
        sortPage.push_back(tempRec);
    }
    runCount++;
    std::sort(sortPage.begin(), sortPage.end(), Comparator(sortorder, false));
    dbPage.EmptyItOut();
    for(int i = 0; i < sortPage.size(); i++) {
        if(dbPage.Append((sortPage[i])) == 0) {
            if(midFile.GetLength() == 0)
            {
                midFile.AddPage(&dbPage, midFile.GetLength());
                pagesInRun[runCount - 1]++;
            }
            else
            {
                midFile.AddPage(&dbPage, midFile.GetLength() - 1);
                pagesInRun[runCount - 1]++;
            }
            dbPage.EmptyItOut();
            dbPage.Append((sortPage[i]));
        }
    }
    // same reason, there must be one page left, and has not been added into the file
    if(midFile.GetLength() == 0)
    {
        midFile.AddPage(&dbPage, midFile.GetLength());
        pagesInRun[runCount - 1]++;
    }
    else
    {
        midFile.AddPage(&dbPage, midFile.GetLength() - 1);
        pagesInRun[runCount - 1]++;
    }
    dbPage.EmptyItOut();
    for(int i = 0; i < sortPage.size(); i++) {
        delete sortPage[i];
    }
    sortPage.clear();
    pageCount = 0;
    midFile.Close();

    midFile.Open(1, filename);
    if(runlen >= midFile.GetLength()) {
        Page tempPage;
        Record tempRec;
        int c = 0;
        while(c < midFile.GetLength() - 1) {
            midFile.GetPage(&tempPage, c);
            while(tempPage.GetFirst(&tempRec) == 1) {
                out->Insert(&tempRec);
            }
            c++;
        }
        midFile.Close();
        out->ShutDown();
    }
    else {
        // construct priority queue over sorted runs and dump sorted data
        // into the out pipe
        typedef priority_queue<Record*, vector<Record*>, Comparator> my_pQueue;
        my_pQueue sorted_runs((Comparator(sortorder, true)));

        // any better way to locate which page the top element in PQ comes from?

        // detemine which page the record comes from in each run
        Page** firstPages = new Page*[runCount];
        for(int i = 0; i < runCount; i++) {
            firstPages[i] = new Page;
        }
        int whichPage = 0;
        for(int i = 0; i < runCount; i++) {
            int t = whichPage;
            if(i == 0) whichPage = 0;
            else whichPage = pagesInRun[i - 1] + t;
            midFile.GetPage(firstPages[i], whichPage);
        }
        // how many pages left in a run
        // can be replaced by vector

        int pagesLeftInRun[10000] = {0};
        for(int i = 0; i < runCount; i++) {
            pagesLeftInRun[i] = pagesInRun[i] - 1;
        }
        // comes to which record in the current page in each run
        Record** firstRecs = new Record*[runCount];
        for(int i = 0; i < runCount; i++) {
            firstRecs[i] = new Record;
        }
        for(int i = 0; i < runCount; i++) {
            firstPages[i]->GetFirst(firstRecs[i]);
        }
        // initialize the priority queue
        for(int i = 0; i < runCount; i++) {
            sorted_runs.push(firstRecs[i]);
        }
        // merge these runs
        whichPage = 0;
        while(!sorted_runs.empty()) {
            int pos = 0;
			
            for(int i = 0; i < runCount; i++) {
                if(sorted_runs.top() == firstRecs[i]) {
                    pos = i;
                    break;
                }
            }
            out->Insert(sorted_runs.top());
            sorted_runs.pop();
            
            if(firstPages[pos]->GetFirst(firstRecs[pos]) == 1) {
                sorted_runs.push(firstRecs[pos]);
                //size = sorted_runs.size();
            }
            else {
                if(pagesLeftInRun[pos] > 0) {
                    whichPage = 0;
                    for(int it = 0; it < pos; it++) {
                        whichPage += pagesInRun[it];
                    }
                    whichPage = whichPage + pagesInRun[pos] - pagesLeftInRun[pos];
                    firstPages[pos]->EmptyItOut();
                    midFile.GetPage(firstPages[pos], whichPage);
                    firstPages[pos]->GetFirst(firstRecs[pos]);
                    sorted_runs.push(firstRecs[pos]);
                    //size = sorted_runs.size();
                    pagesLeftInRun[pos]--;
                }
            }
        }

        midFile.Close();
        // finally shut down the out pipe
        out->ShutDown ();
        delete firstRec;
        for(int i = 0; i < runCount; i++) {
            delete firstRecs[i];
        }
        delete [] firstRecs;
        for(int i = 0; i < runCount; i++) {
            delete firstPages[i];
        }
        delete [] firstPages;
    }
    delete mybqi;
}

BigQ :: BigQ (Pipe &in, Pipe &out, OrderMaker &sortorder, int runlen) {
	//read data from in pipe sort them into runlen pages
	//There won't be pages without records...

	BigQInfo *bqi=new BigQInfo();
	bqi->in=&in;
	bqi->out=&out;
	bqi->order=sortorder;
	bqi->runLength=runlen;
	bqi->midFile=midFile;

	pthread_t BigQthread;

	pthread_create(&BigQthread,NULL,&BigQThread,(void*)bqi);

}

BigQ::~BigQ () {
}
